package org.jeecg.modules.iot.middleware.kafka.consumer;

import com.fasterxml.jackson.core.JsonProcessingException;

import lombok.RequiredArgsConstructor;
import org.jeecg.common.base.BaseMap;
import org.jeecg.common.modules.redis.client.JeecgRedisClient;
import org.jeecg.common.util.RedisUtil;
import org.jeecg.jt808.JTT808Decoder;
import org.jeecg.jt808.modbus.MqttMessageJson;
import org.jeecg.jt808.req.LocationNewMsg;
import org.jeecg.jt808.vo.ClientConnectEvent;


import org.jeecg.modules.iot.cache.EBikeCache;
import org.jeecg.modules.iot.middleware.service.IConsumptionService;
import org.jeecg.modules.iot.middleware.service.ILocationService;
import org.jeecg.modules.iot.middleware.service.IPowerService;
import org.jeecg.modules.iot.middleware.util.GeomUtils;
import org.jeecg.modules.iot.middleware.util.JT808Utils;
import org.jeecg.modules.iot.middleware.util.JsonUtil;
import org.jeecg.modules.iot.middleware.util.LocationUtil;
import org.jeecg.modules.iot.model.FieldStatus;
import org.jeecg.modules.iot.model.Jt808Control.JT808AttributeConverter;
import org.jeecg.modules.iot.model.Location;
import org.jeecg.modules.iot.model.Track.Calories;
import org.jeecg.modules.iot.model.Track.Tracking;
import org.jeecg.modules.iot.model.power.TaiPower;
import org.jeecg.modules.websocket.constant.WsMessageConstant;
import org.jeecg.modules.websocket.enums.WsBussinesType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaOperations;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.DeadLetterPublishingRecoverer;
import org.springframework.kafka.listener.SeekToCurrentErrorHandler;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.util.backoff.FixedBackOff;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.yzh.protocol.t808.T0200;

import java.io.IOException;
import java.math.BigDecimal;
import java.sql.Timestamp;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.time.Instant;
import java.time.ZoneId;
import java.time.ZonedDateTime;
import java.time.format.DateTimeFormatter;
import java.time.format.DateTimeParseException;
import java.util.*;
import java.util.concurrent.CompletableFuture;

@RequiredArgsConstructor
@Component
public class BikeConsumer {

    private static final Logger log = LoggerFactory.getLogger(BikeConsumer.class);

    private final ILocationService locationService;
    private final RedisUtil redisUtil;
    private final IPowerService powerService;
    private final IConsumptionService consumptionService;
    private static final DateTimeFormatter DATE_FORMAT = DateTimeFormatter.ofPattern("yyMMddHHmmss")
            .withZone(ZoneId.of("GMT+8")); // 设置时区
    private final JeecgRedisClient jeecgRedisClient;
    private static final String NATIONAL_STANDARD_SPEED_LIMIT_KEY = "vehicle:%s:nationalStandardSpeedLimit"; //  Redis key 格式
    @Autowired
    private EBikeCache vehicleCache;

    @Value("${jeecg.kafka.location.topic}")
    private String locationTopic;


    @Autowired
    EBikeCache eBikeCache;

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    private final ObjectMapper objectMapper = new ObjectMapper();

    @KafkaListener(topics = "${jeecg.kafka.location.topic}", groupId = "${spring.kafka.consumer.group-id}", concurrency = "12")
    public void listen(String record) throws IOException, ParseException {

        try {
            // 解析位置信息

            MqttMessageJson mqttMessageJson = objectMapper.readValue(record, MqttMessageJson.class);
            /*if (!"865376074560937".equals(mqttMessageJson.getId())){
                return;
            }*/
            LocationNewMsg locationNewMsg = (LocationNewMsg) JTT808Decoder.decode(mqttMessageJson.getData());

            String powerMessage = JsonUtil.createPubJson(WsBussinesType.POWER.getCode(), true, locationNewMsg);

            BaseMap poweParams = new BaseMap();
            poweParams.add("terminalCode", mqttMessageJson.getId());
            poweParams.add("message", powerMessage);
            jeecgRedisClient.sendMessage(WsMessageConstant.EBIKE_REALTIME_MESSAGE, poweParams);

            T0200 msg = (T0200) JT808Utils.decode(mqttMessageJson.getData());
            Map<Integer, Object> s = msg.getAttributes();
            Location location = createLocation(mqttMessageJson, msg);
            if (location != null && locationNewMsg != null) {
                if (location.getRMileage() == 0f) {
                    Integer remainingCapacity = locationNewMsg.getRemainingCapacity();
                    // 确保 remainingCapacity 不为空
                    if (remainingCapacity != null) {
                        location.setRMileage(remainingCapacity * 0.03f);
                    }
                }

                TaiPower taiPower=new TaiPower();
                if (locationNewMsg.getSoh() != null) {
                    taiPower.setSoh(locationNewMsg.getSoh());
                }

                if (locationNewMsg.getRemainingCapacity() != null) {
                    taiPower.setRPow(Float.valueOf(locationNewMsg.getRemainingCapacity()));
                }

                if (locationNewMsg.getSoc() != null) {
                    taiPower.setRPowCap(Float.valueOf(locationNewMsg.getSoc()));
                }

                if (locationNewMsg.getBatteryVoltage() != null) {
                    taiPower.setVoltage(Float.valueOf(locationNewMsg.getBatteryVoltage()));
                }
            }
            pushLocationToFrontEnd(mqttMessageJson, location);

        } catch (Exception e) {

            log.error("解析或处理位置信息失败，record: {}", record, e);
            throw e;
        }
    }


    @KafkaListener(topics = "${jeecg.kafka.location-can.topic}", groupId = "${spring.kafka.consumer.group-id}", concurrency = "12")
    public void listenCan(String record) throws IOException, ParseException {

        try {
            // 解析位置信息

            MqttMessageJson mqttMessageJson = objectMapper.readValue(record, MqttMessageJson.class);
           /* if (!"867156061888089".equals(mqttMessageJson.getId())){
                return;
            }*/

            LocationNewMsg locationNewMsg = (LocationNewMsg) JTT808Decoder.decode(mqttMessageJson.getData());

            String powerMessage = JsonUtil.createPubJson(WsBussinesType.POWER.getCode(), true, locationNewMsg);

            BaseMap poweParams = new BaseMap();
            poweParams.add("terminalCode", mqttMessageJson.getId());
            poweParams.add("message", powerMessage);
            jeecgRedisClient.sendMessage(WsMessageConstant.EBIKE_REALTIME_MESSAGE, poweParams);
            Location location = LocationUtil.createLocationFromLocationNewMsg(mqttMessageJson, locationNewMsg);
            if (location != null) {
                /*更新剩余里程*/
                updateMileage(location, locationNewMsg);
                TaiPower taiPower = createTaiPower(locationNewMsg, location.getDeviceId());
                powerService.add(taiPower);
                eBikeCache.setVehiclePower(mqttMessageJson.getId(),taiPower);
                jeecgRedisClient.sendMessage(WsMessageConstant.EBIKE_REALTIME_MESSAGE, poweParams);
                System.out.println("设备Id对应"+mqttMessageJson.getId()+"电量数据"+taiPower.getRPowCap());
                // 处理 taiPower 逻辑（如果需要）
            }



            String deviceId = mqttMessageJson.getId();


            /*推送位置信息*/
            Map<String, Object> locationData = new HashMap<>();
            locationData.put("latitude", location.getLatitude());
            locationData.put("longitude", location.getLongitude());
            String payload = objectMapper.writeValueAsString(locationData);
            sendKafkaMessage("location",deviceId,payload);
            /*推送剩余里程*/
            Tracking tracking = new Tracking();
            double rMileage = Math.abs(location.getRMileage());
            tracking.setRMileage(roundToTwoDecimalPlaces((float) rMileage));
            CompletableFuture.runAsync(() -> sendMessageToFrontend(deviceId, WsBussinesType.TRACKING, tracking));

            /*推送卡路里*/
            if (locationNewMsg.getCalories()!=null){
                Calories calories=new Calories();
                calories.setCalorise(locationNewMsg.getCalories());
                CompletableFuture.runAsync(() -> sendMessageToFrontend(deviceId, WsBussinesType.CALORIES, calories));
                System.out.println("卡路里数量"+calories.getCalorise());
            }

            /*更新经纬度*/
            vehicleCache.updateVehicleLocation(deviceId, location.getLongitude(), location.getLatitude());
            /*发送车辆状态*/
            FieldStatus fieldStatus = new FieldStatus(location.getStatusField());
            vehicleCache.setFieldStatus(deviceId, fieldStatus);
            /*位置信息存到数据库*/
            addLocationToDatabase(location);
            /*推送位置信息*/
            //String onlineStatus = vehicleCache.getVehicleOnlineStatus(deviceId);
            vehicleCache.setVehicleOnlineStatus(deviceId,"online");
            location.setOnlineStatus("online");
            CompletableFuture.runAsync(() -> sendMessageToFrontend(deviceId, WsBussinesType.LOCATION, location));
            String sportModeRedisKey = String.format(NATIONAL_STANDARD_SPEED_LIMIT_KEY, deviceId);
            CompletableFuture<Boolean> sportModeFuture = CompletableFuture.supplyAsync(() -> (Boolean) redisUtil.get(sportModeRedisKey));
            sportModeFuture.thenAccept(sportModeEnabled -> {
                boolean sportMode = sportModeEnabled != null && sportModeEnabled;
                fieldStatus.setSportMode(sportMode);
                sendMessageToFrontend(deviceId, WsBussinesType.OTHER, fieldStatus);
            });
        } catch (Exception e) {
            log.error("解析或处理位置信息失败，record: {}", record, e);
            throw e;
        }
    }

    @KafkaListener(topics = "${jeecg.kafka.device.status.topic}", groupId = "${spring.kafka.consumer.group-id}", concurrency = "3")
    public void listenDeviceStatus(String record) throws JsonProcessingException {
        try {
            // 解析设备上下线消息
            ClientConnectEvent clientConnectEvent = objectMapper.readValue(record, ClientConnectEvent.class);
            String deviceId = clientConnectEvent.getClientid();
            /*if (!deviceId.equals("867156061919900")){
                return;
            }*/
            String status = "client.connected".equals(clientConnectEvent.getEvent()) ? "online" : "offline";
            vehicleCache.setVehicleOnlineStatus(deviceId, status);
            /*String s=  vehicleCache.getVehicleOnlineStatus(deviceId);
            System.out.println(s);*/
        } catch (JsonProcessingException e) {
            log.error("解析设备上下线消息失败，record: {}", record, e);
            throw e;
        }
    }

    @Bean
    public SeekToCurrentErrorHandler errorHandler(KafkaOperations<Object, Object> template) {
        DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(template);
        return new SeekToCurrentErrorHandler(recoverer, new FixedBackOff(0L, 0L)); // Retry 3 times with 1-second intervals
        // ... 错误处理，失败一次的数据直接发送到死信 ...
    }

    private void processLocation(Location location) {
        // ... 处理位置信息的逻辑 ...sg
        log.info("收到位置信息：{}", location); // 例如，打印位置信息
        // ... 可以将位置信息保存到数据库、进行实时监控等 ...
    }
    // 更新里程数的辅助方法
    private void updateMileage(Location location, LocationNewMsg locationNewMsg) {
        if (location.getRMileage() == 0f) {
            Integer remainingCapacity = locationNewMsg.getRemainingCapacity();
            if (remainingCapacity != null) {
                location.setRMileage(remainingCapacity * 0.03f);
            }
        }
    }

    private Location createLocation(MqttMessageJson mqttMessageJson, T0200 msg) throws ParseException {
        long timestamp;
        try {
            timestamp = Instant.from(DATE_FORMAT.parse(mqttMessageJson.getTimestamp())).toEpochMilli();
        } catch (DateTimeParseException e) {
            log.error("时间戳解析错误：{}", mqttMessageJson.getTimestamp(), e);
            timestamp = Instant.now().toEpochMilli(); // 使用 Instant.now() 获取当前时间戳
        }
        Map<Integer, Object> attributeConverter = msg.getAttributes();
        Integer snnNumber = (Integer) JT808AttributeConverter.convertAttributes(attributeConverter).get("gnssSatelliteCount"); //  链式调用
        Location location = new Location();
        location.setTs(new Timestamp(timestamp)); // 设置时间戳
        location.setDeviceId(mqttMessageJson.getId());
        location.setAlarm(msg.getWarnBit());
        location.setStatusField(msg.getStatusBit());
        location.setLatitude(msg.getLatitude() * 1.0 / 1000000);
        location.setLongitude(msg.getLongitude() * 1.0 / 1000000);
        location.setSnnNumber(snnNumber);
        location.setSpeed((short) msg.getSpeed());
        location.setDirection((short) msg.getDirection());
        location.setElevation((short) msg.getAltitude());
        location.setSeq(mqttMessageJson.getSeq());
        return location;
    }


    private void pushLocationToFrontEnd(MqttMessageJson mqttMessageJson, Location location) {

        String deviceId = mqttMessageJson.getId();
        Tracking tracking = new Tracking();
        double rMileage = Math.abs(location.getRMileage());
        tracking.setRMileage(roundToTwoDecimalPlaces((float) rMileage));
        CompletableFuture.runAsync(() -> sendMessageToFrontend(deviceId, WsBussinesType.TRACKING, tracking));
        vehicleCache.updateVehicleLocation(deviceId, location.getLongitude(), location.getLatitude());
        FieldStatus fieldStatus = new FieldStatus(location.getStatusField());
        vehicleCache.setFieldStatus(deviceId, fieldStatus);
        Location lastLocation = locationService.queryByDeviceId(deviceId);
        calculatePowerConsumptionNew(location, lastLocation);
        Optional.ofNullable(lastLocation).ifPresent(lastLoc -> {
            updateMileage(location, lastLoc);
        });
        addLocationToDatabase(location);
        //String onlineStatus = vehicleCache.getVehicleOnlineStatus(deviceId);
        vehicleCache.setVehicleOnlineStatus( deviceId,"online");
        location.setOnlineStatus("online");
        CompletableFuture.runAsync(() -> sendMessageToFrontend(deviceId, WsBussinesType.LOCATION, location));
        String sportModeRedisKey = String.format(NATIONAL_STANDARD_SPEED_LIMIT_KEY, deviceId);
        CompletableFuture<Boolean> sportModeFuture = CompletableFuture.supplyAsync(() -> (Boolean) redisUtil.get(sportModeRedisKey));
        sportModeFuture.thenAccept(sportModeEnabled -> {
            boolean sportMode = sportModeEnabled != null && sportModeEnabled;
            fieldStatus.setSportMode(sportMode);
            sendMessageToFrontend(deviceId, WsBussinesType.OTHER, fieldStatus);
        });


    }

    // 创建 TaiPower 对象的辅助方法
    private TaiPower createTaiPower(LocationNewMsg locationNewMsg,String deviceId) {
        TaiPower taiPower = new TaiPower();
        taiPower.setTs(new Timestamp(new Date().getTime()));
        taiPower.setDeviceId(deviceId);
        if (locationNewMsg.getSoh() != null) {
            taiPower.setSoh(locationNewMsg.getSoh());
        }
        if (locationNewMsg.getRemainingCapacity() != null) {
            taiPower.setRPow(locationNewMsg.getRemainingCapacity().floatValue());
        }
        if (locationNewMsg.getSoc() != null) {
            taiPower.setRPowCap(locationNewMsg.getSoc().floatValue());
        }
        if (locationNewMsg.getBatteryVoltage() != null) {
            taiPower.setVoltage(locationNewMsg.getBatteryVoltage().floatValue());
        }
        return taiPower;
    }


    private void updateMileage(Location location, Location lastLocation) {
        if (lastLocation == null) {
            return;
        }
        long timeDiff = location.getTs().getTime() - lastLocation.getTs().getTime();
        Float lastMileage = lastLocation.getMileage();
        Float lastDMileage = lastLocation.getDMileage();
        Float lastSMileage = lastLocation.getSMileage();
        Float lastFMileage = lastLocation.getFMileage();

        FieldStatus fieldStatus = new FieldStatus(location.getStatusField());


        if (timeDiff > 60 * 1000 || !fieldStatus.isAcc()) {
            location.setMileage(lastMileage);
            if (!fieldStatus.isAcc()) {
                location.setSMileage(0f);

            } else {
                location.setSMileage(lastLocation.getSMileage());
                location.setFMileage(lastLocation.getFMileage());
            }
            return;
        }


        double distance = GeomUtils.calculateDistance(lastLocation.getLatitude(), lastLocation.getLongitude(),
                location.getLatitude(), location.getLongitude());
        if (distance >= 1000) { // 如果 distance 不超过 1000 米
            distance = 0;
        }
        //单次里程

        float newMileage = (lastMileage == null ? 0f : lastMileage) + (float) distance;
        float newSMileage = (lastSMileage == null ? 0f : lastSMileage) + (float) distance;
        float newFMileage = (lastFMileage == null ? 0f : lastFMileage) + (float) distance;
        location.setMileage(newMileage);
        location.setFMileage(newFMileage);
        location.setSMileage(newSMileage);
    }


    public void calculatePowerConsumptionNew(Location currentLocation, Location lastLocation) {

        if (lastLocation == null) {
            return; // 如果没有上一个位置信息，直接返回
        }
        String deviceId = currentLocation.getDeviceId();
        TaiPower fPower = powerService.queryLastByDeviceId(deviceId);
        if (lastLocation != null) {

            if (fPower == null) {
                return; // 处理 fPower 为 null 的情况
            }

            double distance = GeomUtils.calculateDistance(lastLocation.getLatitude(), lastLocation.getLongitude(),
                    currentLocation.getLatitude(), currentLocation.getLongitude());

            // 里程以米为单位进行计算
            float newMileage = (lastLocation.getMileage() == null ? 0f : lastLocation.getMileage()) + (float) distance;
            float newSMileage = (lastLocation.getSMileage() == null ? 0f : lastLocation.getSMileage()) + (float) distance;
            float newFMileage = (lastLocation.getFMileage() == null ? 0f : lastLocation.getFMileage()) + (float) distance;

            currentLocation.setMileage(newMileage);
            currentLocation.setSMileage(newSMileage);
            currentLocation.setFMileage(newFMileage);


            float rmilege = 0f;
            float avrConsum = consumptionService.calculateAverageEConsum(deviceId, 10);

            if (avrConsum != 0) {
                //rmilege = fPower.getRPow() / avrConsum;
                rmilege = fPower.getRPow() / avrConsum;

            }
            currentLocation.setRMileage(rmilege);
        }
    }


    private float roundToTwoDecimalPlaces(float value) {
        return BigDecimal.valueOf(value).setScale(2, BigDecimal.ROUND_HALF_UP).floatValue();
    }

    private void addLocationToDatabase(Location location) {
        CompletableFuture.runAsync(() -> locationService.add(location)); // 异步保存到数据库
    }

    private void sendMessageToFrontend(String deviceId, WsBussinesType type, Object data) {
        // 缓存 WsBussinesType.getCode() 结果
        int code = type.getCode();
        String message = JsonUtil.createPubJson(code, true, data); // 避免重复调用 type.getCode()

        BaseMap params = new BaseMap();
        params.add("terminalCode", deviceId);
        params.add("message", message);
        jeecgRedisClient.sendMessage(WsMessageConstant.EBIKE_REALTIME_MESSAGE, params);
    }

    private void recordProcessingTime(long processingTime) {
        try {
            log.info("位置数据处理时间：{} 毫秒", processingTime);
        } catch (Exception e) {
            log.error("记录处理时间失败", e);
            // 处理异常
        }
    }

    public void sendKafkaMessage(String topic, String key, String payload) {
        String defaultKey = "default_key"; // 定义默认的 key
        String actualKey = (key != null && !key.isEmpty()) ? key : defaultKey; // 使用 key 或默认 key
        ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(topic, actualKey, payload);
        future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
            @Override
            public void onSuccess(SendResult<String, String> stringStringSendResult) {
            }

            @Override
            public void onFailure(Throwable ex) {
            }
        });
    }

}